package cn.getech.data.development.job.util

import java.sql.{Connection, DriverManager, Statement}
import java.util.Properties

import org.apache.spark.sql.{DataFrame, SparkSession}

object SparkJobUtils {

  def getSparkSession(jobName:String):SparkSession={
    val spark = SparkSession.builder().appName(jobName).enableHiveSupport().getOrCreate()
    spark
  }

  /**
   * 查询hive表加密的字段
   * @param spark
   * @param hiveDatabase
   * @param hiveTables
   */
  def queryEncryptedField(spark:SparkSession,hiveDatabase:String,hiveTables:String*){
    val properties: Properties = PropertiesUtils.getProperties("common.properties")

    val driver: String = PropertiesUtils.getValue(properties,"mysql_driver")
    val url: String = PropertiesUtils.getValue(properties,"mysql_url")
    val user: String = PropertiesUtils.getValue(properties,"mysql_user")
    val passwd: String = PropertiesUtils.getValue(properties,"mysql_passwd")

    val prop = new Properties()
    prop.put("driver", driver)
    prop.put("user", user)
    prop.put("password", passwd)

    val sqlSB = new StringBuilder("select t1.`ency_style_id`, t1.`table_name`, t1.`field_name` ")
    sqlSB.append("from ( select ")
      .append("f.`table_id` as table_id, f.`ency_style_id` as ency_style_id, t.`table_name` as table_name, f.`field_name` as field_name ")
      .append("from `bdp-admin`.`bdp_table_field` f inner join ( ")
      .append("select table_name, id from `bdp-admin`.`bdp_table_info` ")
      .append(" where table_name in ( ")

    for (hiveTable <- hiveTables) {
      sqlSB.append(hiveDatabase).append(".`").append(hiveTable).append("`,")
    }
    sqlSB.deleteCharAt(sqlSB.length-1).append(")) t on f.table_id = t.id) t1 where t1.ency_style_id = 29")

    val frame: DataFrame = spark.read.jdbc(url,s"(${sqlSB.toString()}) as m",prop)
    frame.select("field_name").collect()

  }

  /**
   * 创建HiveServer2链接
   * @param url hiveserver2 url
   * @param driver hive驱动类
   * @param user hive用户名
   * @param passwd hive passwd
   * @return Statement
   */
  def connectHiveServer2(url:String,driver:String,user:String,passwd:String): Statement ={
    Class.forName(driver)
    val connection: Connection = DriverManager.getConnection(url,user,passwd)
    val statement: Statement = connection.createStatement()
    statement
  }

}
